package org.myfram.flink.flinkonjar.common.connector.oracle.sink;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.myfram.flink.flinkonjar.common.properties.OracleProperties;

import java.sql.PreparedStatement;

public class OracleSinkBuilder {

    public static <T> SinkFunction<T> build(
            OracleProperties properties,
            String insertSql,
            JdbcStatementBuilder<T> builder
    ) {
        return JdbcSink.sink(
                insertSql,
                builder,
                JdbcExecutionOptions.builder()
                        .withBatchSize(properties.getBatchSize())
                        .withBatchIntervalMs(properties.getBatchIntervalMs())
                        .withMaxRetries(properties.getMaxRetries())
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(properties.getDriver())
                        .withUrl(properties.getUrl())
                        .withUsername(properties.getUsername())
                        .withPassword(properties.getPassword())
                        .build()
        );
    }
}
